package com.edu.yx.app.App_03_DwdApp.Dwd_03_Trade;

import com.edu.yx.app.APP_01_BaseApp.BaseSqlApp;
import com.edu.yx.common.Constant;
import com.edu.yx.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_02_Trade_Cart_AddDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_02_Trade_Cart_AddDetail().init(
                3032,
                2,
                "Dwd_02_Trade_Cart_AddDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        //1.先拿ods_db的原始数据
        readOdsDb(tEnv,"Dwd_02_Trade_Cart_AddDetail");


        //2.由于这里只需要有加购行为就行
        Table result = tEnv.sqlQuery("select " +
                "data['id'] cart_id, " +
                "data['user_id'] user_id," +
                "type, " +
                "ts " +
                " from ods_db  " +
                " where `database`='edu'  " +
                " and `table`='cart_info'  " +
                " and `type` = 'insert'  "
        );
//         result.execute().print();


        tEnv.executeSql("create table dwd_trade_cart_add(" +
                "cart_id string," +
                "user_id string," +
                "type string," +
                "ts bigint" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_CART_ADD)
        );

        //把结果插入到表dwd_trade_cart_add
        result.executeInsert("dwd_trade_cart_add");
    }


}
